/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.cluster.impl.kryo;

import java.io.IOException;
import java.net.InetAddress;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.sabayframework.activity.Activity;
import org.sabayframework.activity.ActivityMonitor;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.impl.NullChannel;
import org.sabayframework.cluster.Cluster;
import org.sabayframework.cluster.ClusterEvent;
import org.sabayframework.cluster.ClusterListener;
import org.sabayframework.cluster.Computer;
import org.sabayframework.cluster.DistributedActivity;
import org.sabayframework.cluster.NodeExecutionQueue;
import org.sabayframework.cluster.impl.ComputerIdImpl;
import org.sabayframework.log.Log;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryonet.Server;

/**
 * Server for processing remote activity requests.
 * @author Alex Antonau
 * */
public class KryoServer	implements ClusterListener, NodeExecutionQueue {

	private final int port;
	private final Server server;
	
	private final Cluster cluster;
	private final Computer localComputer;
	
    /** Main lock guarding all access */
    private final ReentrantLock lock;
    /** Condition for waiting takes */
    private final Condition notEmpty;
    
    private final RequestManager requestManager;
    private final NetworkManager networkManager;
    private final KryoListener kryoListener;
    
    private volatile boolean active;
    private ActivityMonitor<?, ?> rmaMonitor;
        
	public KryoServer(Cluster cluster, Computer localComputer, int port, int maxBufferSize, int maxQueueSize) {
		this.port = port;
		this.cluster = cluster;
		this.localComputer = localComputer;
        this.lock = new ReentrantLock();
        this.notEmpty = lock.newCondition();
        
        server = new Server(maxBufferSize, maxBufferSize);
        final Kryo kryo = server.getKryo();
        initKryo(kryo);
        requestManager = new RequestManager(kryo, lock, localComputer.getComputerId(), maxQueueSize, maxBufferSize);
        kryoListener = new KryoListener(requestManager, localComputer.getExecutionQueue(), kryo);
        networkManager = new NetworkManager(kryoListener, server, localComputer.getComputerId());
        requestManager.setNetworkManager(networkManager);
        kryoListener.setNetworkManager(networkManager);
	}
	
	public int getPort() {
		return port;
	}

	public void start() throws IOException{
		lock.lock();
		try{
			Log.debug("KryoServer.start on ", localComputer, ":" ,port);
			active = true;
			server.addListener(kryoListener);
			server.bind(port);
			server.start();
			cluster.addEventListener(this);
			rmaMonitor = localComputer.getExecutionQueue().submit(new RequestManagementActivity(requestManager), NullChannel.getInstance(), NullChannel.getInstance());
		} finally{
			lock.unlock();
		}
	}
	@Override
	public void stop(){
		lock.lock();
		try{
			if(active){
				Log.debug("KryoServer.stop on ", localComputer, ":" ,port);
				active = false;
				requestManager.failAllRequests();
				notEmpty.signalAll();
				cluster.removeEventListener(this);
				server.removeListener(kryoListener);
				server.stop();
				rmaMonitor.kill(new RuntimeException("Stop request management activity"));
				rmaMonitor = null;
			}
		} finally{
			lock.unlock();
		}
	}
	
	@Override
	public void handleEvent(ClusterEvent event) {}

	@Override
	public void computerJoined(Computer c) {}

	@Override
	public void computerLeft(Computer c) {
		requestManager.removeRequests(c);
		networkManager.removeConnections(c.getId());
	}
	
	@Override
	public <I, O> ActivityMonitor<I,O> submit(Activity<I, O> activity,	Channel<I> input, Channel<O> output) {
		return localComputer.getExecutionQueue().submit(
				activity, 
				input, 
				output);
	}

	@Override
	public <I, O> ActivityMonitor<I, O> execute(Activity<I, O> activity,
			Channel<I> input, Channel<O> output) {
		return localComputer.getExecutionQueue().execute(
				activity, 
				input, 
				output);
	}

	@SuppressWarnings("unchecked")
	@Override
	public <I, O> ActivityMonitor<I, O> submit(
					DistributedActivity<I, O> da, 
					Channel<I> input, 
					Channel<O> output) {
		DistributedActivityContext r = new DistributedActivityContext();
		r.setDistributedActivity((DistributedActivity<Object, Object>) da);
		r.setInput((Channel<Object>) input);
		r.setOutput((Channel<Object>) output);
		r.setClientId(localComputer.getComputerId());
		Request req = new Request(RequestType.START, r);
		requestManager.submitRequest(req);
		return (ActivityMonitor<I, O>) r;
	}

	@Override
	public int workerCount() {
		return cluster.getChildrenCount();
	}

	@Override
	public int size() {
		return 0;
	}

	@Override
	public boolean isEmpty() {
		return true;
	}

	@Override
	public void channelChanged(Channel<?> input) {
		localComputer.getExecutionQueue().channelChanged(input);
	}

	public static void initKryo(Kryo kryo){
		kryo.setRegistrationRequired(false);
		kryo.register(RequestType.class);
		kryo.register(NetworkPacket.class);
		kryo.register(InetAddress.class);
		kryo.register(ComputerIdImpl.class);
	}

	public Cluster getCluster() {
		return cluster;
	}

	public Computer getLocalComputer() {
		return localComputer;
	}

	public RequestManager getRequestManager() {
		return requestManager;
	}

	public NetworkManager getNetworkManager() {
		return networkManager;
	}

	public KryoListener getKryoListener() {
		return kryoListener;
	}
	
	public Kryo getKryo(){
		return server.getKryo();
	}

}
